package controller;

import operators.UV_AllWindowFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import pojo.UserBehavior;

import java.net.URL;

/*
* 2021-05-24 16:24:55
* 计算uv
* */
public class UniqueVisitor {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        URL resource = UniqueVisitor.class.getResource("/UserBehavior.csv");
        DataStream<String> fileDataStream = env.readTextFile(resource.getPath());

        DataStream<UserBehavior> watermarkDataStream = fileDataStream.map(line -> {
            String[] fields = line.split(",");
            return new UserBehavior(Long.valueOf(fields[0]), Long.valueOf(fields[1]), Integer.valueOf(fields[2]), fields[3], Long.valueOf(fields[4]));
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserBehavior>noWatermarks()
                        .withTimestampAssigner((SerializableTimestampAssigner<UserBehavior>) (element, recordTimestamp) -> element.getTimestamp() * 1000L)
        );

        watermarkDataStream.filter(data -> "pv".equals(data.getBehavior()))
                .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
                .apply(new UV_AllWindowFunction())
                .print();

        env.execute();
    }
}
